package drds.binlog.metadata;

import com.google.common.base.Function;
import com.google.common.collect.ComputingMap;
import com.google.common.collect.Lists;
import drds.binlog.common.position.ClientInfo;
import drds.binlog.common.position.LogPosition;
import drds.binlog.common.position.Position;
import drds.binlog.common.utils.JsonUtils;
import org.apache.commons.io.FileUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.MDC;
import org.springframework.util.Assert;

import java.io.File;
import java.io.IOException;
import java.nio.charset.Charset;
import java.util.*;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * 基于文件刷新的metaManager实现
 * <p>
 * <pre>
 * 策略：
 * 1. 先写内存，然后定时刷新数据到File
 * 2. 数据采取overwrite模式(只保留最后一次)，通过logger实施append模式(记录历史版本)
 * </pre>
 */
public class FileMixedMetaDataManager extends MemoryMetaDataManager implements MetaDataManager
{

    private static final Logger logger = LoggerFactory.getLogger(FileMixedMetaDataManager.class);
    private static final Charset charset = Charset.forName("UTF-8");
    @SuppressWarnings("serial")
    private final Position nullPosition = new Position()
    {
    };
    private File dataDir;
    private String dataFileName = "metadata.data";
    private Map<String, File> destinationToFileMap;
    private ScheduledExecutorService scheduledExecutorService;
    private long period = 1000;                                               // 单位ms
    private Set<ClientInfo> clientInfoSet;//需要进行更新的

    public void start()
    {
        super.start();
        Assert.notNull(dataDir);
        if (!dataDir.exists())
        {
            try
            {
                FileUtils.forceMkdir(dataDir);
            } catch (IOException e)
            {
                throw new MetaDataManagerException(e);
            }
        }

        if (!dataDir.canRead() || !dataDir.canWrite())
        {
            throw new MetaDataManagerException("dir[" + dataDir.getPath() + "] can not read/write");
        }

        destinationToFileMap = ComputingMap.makeComputingMap(new Function<String, File>()
        {

            public File apply(String destination)
            {
                return getDataFile(destination);
            }
        });

        scheduledExecutorService = Executors.newScheduledThreadPool(1);
        destinationToClientInfoListMap = ComputingMap.makeComputingMap(new Function<String, List<ClientInfo>>()
        {

            public List<ClientInfo> apply(String destination)
            {
                return getClientInfoList1(destination);
            }
        });

        clientInfoToPositionMap = ComputingMap.makeComputingMap(new Function<ClientInfo, Position>()
        {

            public Position apply(ClientInfo clientInfo)
            {
                Position position = loadPosition(clientInfo.getDestination(), clientInfo);
                if (position == null)
                {
                    return nullPosition; // 返回一个空对象标识，避免出现异常
                } else
                {
                    return position;
                }
            }
        });

        clientInfoSet = Collections.synchronizedSet(new HashSet<ClientInfo>());

        // 启动定时工作任务
        scheduledExecutorService.scheduleAtFixedRate(new Runnable()
                                                     {

                                                         public void run()
                                                         {
                                                             List<ClientInfo> clientInfoList = new ArrayList<ClientInfo>(clientInfoSet);
                                                             for (ClientInfo clientInfo : clientInfoList)
                                                             {
                                                                 MDC.put("destination", String.valueOf(clientInfo.getDestination()));
                                                                 try
                                                                 {
                                                                     // 定时将内存中的最新值刷到file中，多次变更只刷一次
                                                                     if (logger.isInfoEnabled())
                                                                     {
                                                                         LogPosition logPosition = (LogPosition) getPosition(clientInfo);
                                                                         logger.info("clientId:{} cursor:[{},{},{},{},{}] address[{}]", new Object[]{
                                                                                 clientInfo.getClientId(), logPosition.getEntryPosition().getJournalName(),
                                                                                 logPosition.getEntryPosition().getPosition(), logPosition.getEntryPosition().getTimestamp(),
                                                                                 logPosition.getEntryPosition().getServerId(), logPosition.getEntryPosition().getGtid(),
                                                                                 logPosition.getLogIdentity().getSocketAddress().toString()});
                                                                     }
                                                                     flush(clientInfo.getDestination());
                                                                     clientInfoSet.remove(clientInfo);
                                                                 } catch (Throwable e)
                                                                 {
                                                                     // ignore
                                                                     logger.error("period update" + clientInfo.toString() + " curosr failed!", e);
                                                                 }
                                                             }
                                                         }
                                                     },
                period,
                period,
                TimeUnit.MILLISECONDS);
    }

    public void stop()
    {
        super.stop();

        flush();// 刷新数据
        scheduledExecutorService.shutdownNow();
        destinationToClientInfoListMap.clear();
        clientInfoToBatchIdToPositionRangeMapMap.clear();
    }

    //s
    public void subscribe(final ClientInfo clientInfo) throws MetaDataManagerException
    {
        super.subscribe(clientInfo);

        // 订阅信息频率发生比较低，不需要做定时merge处理
        scheduledExecutorService.submit(new Runnable()
        {

            public void run()
            {
                flush(clientInfo.getDestination());
            }
        });
    }

    public void unsubscribe(final ClientInfo clientInfo) throws MetaDataManagerException
    {
        super.unsubscribe(clientInfo);

        // 订阅信息频率发生比较低，不需要做定时merge处理
        scheduledExecutorService.submit(new Runnable()
        {

            public void run()
            {
                flush(clientInfo.getDestination());
            }
        });
    }

    public void updatePosition(ClientInfo clientInfo, Position position) throws MetaDataManagerException
    {
        clientInfoSet.add(clientInfo);// 添加到任务队列中进行触发
        super.updatePosition(clientInfo, position);
    }

    public Position getPosition(ClientInfo clientInfo) throws MetaDataManagerException
    {
        Position position = super.getPosition(clientInfo);
        if (position == nullPosition)
        {
            return null;
        } else
        {
            return position;
        }
    }

    //

    private File getDataFile(String destination)
    {
        File file = new File(dataDir, destination);
        if (!file.exists())
        {
            try
            {
                FileUtils.forceMkdir(file);
            } catch (IOException e)
            {
                throw new MetaDataManagerException(e);
            }
        }

        return new File(file, dataFileName);
    }

    private FileMetaDataClientInfoAndLogPositionListContext getFileMetaDataClientInfoAndLogPositionListContext(File file)
    {
        try
        {
            if (!file.exists())
            {
                return null;
            }

            String json = FileUtils.readFileToString(file, charset.name());
            return JsonUtils.unmarshalFromString(json, FileMetaDataClientInfoAndLogPositionListContext.class);
        } catch (IOException e)
        {
            throw new MetaDataManagerException(e);
        }
    }

    private void flush()
    {
        for (String destination : destinationToClientInfoListMap.keySet())
        {
            flush(destination);
        }
    }

    private void flush(String destination)
    {
        flush(destination, destinationToFileMap.get(destination));
    }

    private void flush(String destination, File file)
    {
        FileMetaDataClientInfoAndLogPositionListContext fileMetaDataClientInfoAndLogPositionListContext = new FileMetaDataClientInfoAndLogPositionListContext();
        if (destinationToClientInfoListMap.containsKey(destination))
        {
            synchronized (destination.intern())
            { // 基于destination控制一下并发更新
                fileMetaDataClientInfoAndLogPositionListContext.setDestination(destination);

                List<FileMetaDataClientInfoAndLogPosition> fileMetaDataClientInfoAndLogPositionList = Lists.newArrayList();
                List<ClientInfo> clientInfoList = destinationToClientInfoListMap.get(destination);
                for (ClientInfo clientInfo : clientInfoList)
                {
                    FileMetaDataClientInfoAndLogPosition fileMetaDataClientInfoAndLogPosition = new FileMetaDataClientInfoAndLogPosition();
                    fileMetaDataClientInfoAndLogPosition.setClientInfo(clientInfo);
                    Position position = clientInfoToPositionMap.get(clientInfo);
                    if (position != null && position != nullPosition)
                    {
                        fileMetaDataClientInfoAndLogPosition.setLogPosition((LogPosition) position);
                    }

                    fileMetaDataClientInfoAndLogPositionList.add(fileMetaDataClientInfoAndLogPosition);
                }

                fileMetaDataClientInfoAndLogPositionListContext.setFileMetaDataClientInfoAndLogPositionList(fileMetaDataClientInfoAndLogPositionList);
            }

            String json = JsonUtils.marshalToString(fileMetaDataClientInfoAndLogPositionListContext);
            try
            {
                FileUtils.writeStringToFile(file, json);
            } catch (IOException e)
            {
                throw new MetaDataManagerException(e);
            }
        }
    }

    private List<ClientInfo> getClientInfoList1(String destination)
    {
        List<ClientInfo> clientInfoList = Lists.newArrayList();

        FileMetaDataClientInfoAndLogPositionListContext fileMetaDataClientInfoAndLogPositionListContext = getFileMetaDataClientInfoAndLogPositionListContext(destinationToFileMap.get(destination));
        if (fileMetaDataClientInfoAndLogPositionListContext == null)
        {
            return clientInfoList;
        }

        List<FileMetaDataClientInfoAndLogPosition> fileMetaDataClientInfoAndLogPositionList = fileMetaDataClientInfoAndLogPositionListContext.getFileMetaDataClientInfoAndLogPositionList();
        if (fileMetaDataClientInfoAndLogPositionList == null)
        {
            return clientInfoList;
        }

        for (FileMetaDataClientInfoAndLogPosition fileMetaDataClientInfoAndLogPosition : fileMetaDataClientInfoAndLogPositionList)
        {
            if (fileMetaDataClientInfoAndLogPosition.getClientInfo().getDestination().equals(destination))
            {
                clientInfoList.add(fileMetaDataClientInfoAndLogPosition.getClientInfo());
            }
        }

        return clientInfoList;
    }

    private Position loadPosition(String destination, ClientInfo clientInfo)
    {
        FileMetaDataClientInfoAndLogPositionListContext fileMetaDataClientInfoAndLogPositionListContext = getFileMetaDataClientInfoAndLogPositionListContext(destinationToFileMap.get(destination));
        if (fileMetaDataClientInfoAndLogPositionListContext == null)
        {
            return null;
        }

        List<FileMetaDataClientInfoAndLogPosition> fileMetaDataClientInfoAndLogPositionList = fileMetaDataClientInfoAndLogPositionListContext.getFileMetaDataClientInfoAndLogPositionList();
        if (fileMetaDataClientInfoAndLogPositionList == null)
        {
            return null;
        }

        for (FileMetaDataClientInfoAndLogPosition fileMetaDataClientInfoAndLogPosition : fileMetaDataClientInfoAndLogPositionList)
        {
            if (fileMetaDataClientInfoAndLogPosition.getClientInfo() != null && fileMetaDataClientInfoAndLogPosition.getClientInfo().equals(clientInfo))
            {
                return fileMetaDataClientInfoAndLogPosition.getLogPosition();
            }
        }

        return null;
    }

    public void setDataDir(String dataDir)
    {
        this.dataDir = new File(dataDir);
    }

    public void setDataDirByFile(File dataDir)
    {
        this.dataDir = dataDir;
    }

    public void setPeriod(long period)
    {
        this.period = period;
    }

}
